package com.example.flow

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.asFlow
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.first
import kotlinx.coroutines.flow.flatMapConcat
import kotlinx.coroutines.flow.flatMapLatest
import kotlinx.coroutines.flow.flatMapMerge
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.flowOf
import kotlinx.coroutines.flow.fold
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.onEach
import kotlinx.coroutines.flow.reduce
import kotlinx.coroutines.flow.single
import kotlinx.coroutines.flow.take
import kotlinx.coroutines.flow.toList
import kotlinx.coroutines.flow.toSet
import kotlinx.coroutines.flow.transform
import kotlinx.coroutines.flow.zip
import kotlinx.coroutines.runBlocking
import org.junit.Test

class CoroutineTest02 {

    suspend fun performRequest(request: Int): String {
        return "Response $request"
    }

    /**
     * 转换符 map transform
     */
    @Test
    fun `test transform flow operator`() = runBlocking<Unit> {
//        (1..3).asFlow().map { request->performRequest(request) }
//            .collect { value -> println("value=$value") }

        (1..3).asFlow().transform {
            emit("one $it")
            emit("two $it")
            emit(performRequest(it))
        }.collect { value -> println("value=$value") }
    }

    fun nubmbers() = flow<Int> {
        try {
            emit(1)
            emit(2)
            println("This line will not execute")
            emit(3)
        } finally {
            println("Finally in numbers")
        }
    }

    @Test
    fun `test limit length operator`() = runBlocking<Unit> {
        //限制长度
        nubmbers().take(2).collect { value -> println(value) }
    }


    @Test
    fun `test terminal operator`() = runBlocking<Unit> {
        val sum = (1..2).asFlow()
            .map { it * it }
//            .first { true }
//            .single()
            .fold(1) { a, value ->
                a * value
            }
//            .reduce { a, b -> a + b }
//            .toSet()
//            .toList()
        println(sum)
    }

    /**
     * 组合
     */
    @Test
    fun `test zip`() = runBlocking<Unit> {
        //number 和 strs 是异步的，等待那个最久的完成才可以组合输出
        val number = (1..3).asFlow().onEach { delay(300) }
        val strs = flowOf("One", "Two", "Three").onEach { delay(400) }
        val startTime = System.currentTimeMillis()
        number.zip(strs) { a, b -> "a=$a b=$b" }.collect {
            println("$it at ${System.currentTimeMillis() - startTime} ms")
        }
    }


    fun requestFlow(i: Int) = flow<String> {
        emit("$i First")
        delay(500)
        emit("$i Second")
//        delay(500)
//        emit("$i Three")
    }

    /**
     * 展平模式
     */
    @Test
    fun `test flatMapContact`() = runBlocking<Unit> {
        val startTime = System.currentTimeMillis()
        (1..3).asFlow()
            .onEach { delay(100) }
//            .map { requestFlow(it) }
            .flatMapConcat { requestFlow(it) }
            .collect { println("$it at ${System.currentTimeMillis() - startTime} ms") }

    }

    /**
     * 合并模式
     */
    @Test
    fun `test flatMapMerge`() = runBlocking<Unit> {
        val startTime = System.currentTimeMillis()
        (1..3).asFlow()
            .onEach { delay(100) }
//            .map { requestFlow(it) }
            .flatMapMerge { requestFlow(it) }//针对emit归类
            .collect { println("$it at ${System.currentTimeMillis() - startTime} ms") }
    }

    @Test
    fun `test flatMapLatest`() = runBlocking<Unit> {
        val startTime = System.currentTimeMillis()
        (1..3).asFlow()
            .onEach { delay(100) }
//            .map { requestFlow(it) }
            .flatMapLatest { requestFlow(it) }
//        1 First at 138 ms
//        2 First at 291 ms
//        3 First at 392 ms
//        3 Second at 894 ms 只要最后的emit，其他的都是直接到最后一个输出
            .collect { println("$it at ${System.currentTimeMillis() - startTime} ms") }
    }

}